import time
import threading

import aiohttp
import asyncio

from core.redis_server import RedisServer
from settings import THREADING_POOL_NUMBER, PROXY_TESTER_REQUEST_INTERVAL, PROXY_TESTER_REQUEST_NUMBER, \
    EXPIRATION_TESTER_INTERVAL, SOURCE_TESTER_INTERVAL, PROXY_TYPE, USERNAME, PWD


class ProxyTester(object):

    def __init__(self):
        self._loop = None
        self.redis = RedisServer()

    async def _send(self, proxy_info):      # proxy_info = {'proxy': '127.0.0.1', 'expiration': '时间戳'}
        async with aiohttp.TCPConnector(ssl=False) as connector:
            async with aiohttp.ClientSession(connector=connector) as session:
                try:
                    if PROXY_TYPE == 1:
                        real_proxy = 'http://' + proxy_info['proxy']
                        async with await session.get(url=self.url, headers=self._headers, proxy=real_proxy, timeout=10) as response:
                            if response.status == 200:
                                return proxy_info
                    else:
                        proxy_auth = aiohttp.BasicAuth(USERNAME, PWD)
                        real_proxy = 'http://' + proxy_info['proxy']
                        async with await session.get(url=self.url, headers=self._headers, proxy=real_proxy, timeout=10, proxy_auth=proxy_auth, allow_redirects=False) as response:
                            if response.status == 200:
                                return proxy_info
                except Exception as e:
                    print('host:{}   proxy:{}   error:{}'.format(self.netloc, proxy_info, e))
                    return None

    def _create_task(self, proxies):
        tasks = []
        for proxy in proxies:
            task = asyncio.ensure_future(self._send(proxy))
            task.add_done_callback(self._save_redis)
            tasks.append(task)
        self._loop.run_until_complete(asyncio.wait(tasks))

    def _save_redis(self, future):
        if future.result() is None:
            return
        self.redis.push_source_proxy(self.netloc, future.result())

    def run(self, host, thread_loop):
        self._loop = thread_loop
        asyncio.set_event_loop(thread_loop)
        self.netloc = host
        self.url = 'https://' + host
        proxy_list = self.redis.get_all_proxy_from_interface()
        print(f'代理检测器开启，当前检测HOST为：{self.url}，正在检测...')
        for i in range(0, len(proxy_list), PROXY_TESTER_REQUEST_NUMBER):
            test_proxy = proxy_list[i: i+PROXY_TESTER_REQUEST_NUMBER]
            self._create_task(test_proxy)
            time.sleep(PROXY_TESTER_REQUEST_INTERVAL)

    @property
    def _headers(self):
        return {
            'User-Agent': "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2227.1 Safari/537.36"
        }


class ExpirationTester(object):

    def __init__(self):
        self.redis = RedisServer()

    def run(self):
        print('开始检测代理过期时间...')
        while True:
            self.begin_test()
            time.sleep(EXPIRATION_TESTER_INTERVAL)

    def begin_test(self):
        host_list = self.redis.get_source_list()
        for host in host_list:
            proxy_info_list = self.redis.get_all_proxy_from_source(host)
            for proxy_dict in proxy_info_list:
                if proxy_dict.get('expiration', None) is not None:
                    if proxy_dict['expiration'] <= time.time():
                        self.delete(host, proxy_dict)

    def delete(self, host, proxy_info):
        self.redis.delete_one_proxy_from_source(host, proxy_info)


class SourceTester(object):

    def __init__(self):
        self.redis = RedisServer()
        self.signal = threading.BoundedSemaphore(THREADING_POOL_NUMBER)

    def run(self):
        print('开始监听源队列...')
        self.listen()

    def listen(self):
        # 获得所有的源，检测代理池中的可用代理，并放入源中
        while True:
            source_list = set(self.redis.get_source_list())
            source_name_list = set(self.redis.get_source_list_name())
            # 删除源队列中不存在,而Redis中存在的队列
            x = source_name_list - source_list
            for name in x:
                self.redis.delete_source_list_from_key(name)

            if len(source_list) == 0:
                continue

            for source in source_list:
                # 删除源列表中的空字符
                if len(source) == 0:
                    self.redis.delete_source(source)
                    continue
                # 为每一个线程创建单独loop对象
                thread_loop = asyncio.new_event_loop()
                threading.Thread(target=self.run_proxy_tester, args=(source, thread_loop)).start()
            # 等待所有线程执行完毕
            while True:
                if threading.activeCount() == 1:
                    if len(source_list) <= 5:
                        time.sleep(5)
                    else:
                        time.sleep(SOURCE_TESTER_INTERVAL)
                    break

    def run_proxy_tester(self, host, thread_loop):
        self.signal.acquire()
        tester = ProxyTester()
        tester.run(host, thread_loop)
        self.signal.release()


class ProxyApplyTester(object):
    '''
    主要针对无过期时间的代理的检测
    '''
    def __init__(self):
        self.redis = RedisServer()
        self.signal = threading.BoundedSemaphore(THREADING_POOL_NUMBER)

    def run(self):
        print('开始检测不可用代理')
        while True:
            source_list = self.redis.get_source_list()
            for source in source_list:
                proxy_list = self.redis.get_all_proxy_from_source(source)
                event_loop = asyncio.new_event_loop()
                threading.Thread(target=self.run_proxy_tester, args=(source, event_loop, proxy_list)).start()
                # 等待所有线程执行完毕
                while True:
                    if threading.activeCount() == 1:
                        time.sleep(SOURCE_TESTER_INTERVAL)
                        break

    def run_proxy_tester(self, source, loop, proxy_list):
        self.signal.acquire()
        self.create_task(source, loop, proxy_list)
        self.signal.release()

    def create_task(self, source, loop, proxy_list):
        asyncio.set_event_loop(loop)
        tasks = []
        for proxy in proxy_list:
            if not proxy.get('expiration', None):
                task = asyncio.ensure_future(self.tester(source, proxy))
                task.add_done_callback(self.handle_redis)
                tasks.append(task)
        try:
            loop.run_until_complete(asyncio.wait(tasks))
        except ValueError:
            return

    async def tester(self, source, proxy):
        async with aiohttp.TCPConnector(ssl=False) as connector:
            async with aiohttp.ClientSession(connector=connector) as session:
                try:
                    if PROXY_TYPE == 1:
                        real_proxy = 'http://' + proxy['proxy']
                        async with await session.get(url='https://'+source, headers=self._headers, proxy=real_proxy, timeout=10) as response:
                            if response.status != 200:
                                return (source, proxy)
                    else:
                        proxy_auth = aiohttp.BasicAuth(USERNAME, PWD)
                        real_proxy = 'http://' + proxy['proxy']
                        async with await session.get(url='https://'+source, headers=self._headers, proxy=real_proxy, timeout=10, proxy_auth=proxy_auth, allow_redirects=False) as response:
                            if response.status != 200:
                                return (source, proxy)
                except Exception as e:
                    return (source, proxy)

    def handle_redis(self, future):
        if future.result() is None:
            return
        print('代理{}在域名下{}不可用，删除'.format(future.result()[1], future.result()[0]))
        self.redis.delete_one_proxy_from_source(future.result()[0], future.result()[1])

    @property
    def _headers(self):
        return {
            'User-Agent': "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2227.1 Safari/537.36"
        }


exp_tester = ExpirationTester()
source_tester = SourceTester()
proxy_tester = ProxyApplyTester()
